package flexserverlib.jms.spring.adapters;

import flex.messaging.FlexContext;
import flex.messaging.config.ConfigMap;
import flex.messaging.config.ConfigurationException;
import flex.messaging.messages.AsyncMessage;
import flex.messaging.messages.Message;
import flex.messaging.messages.MessagePerformanceInfo;
import flex.messaging.messages.MessagePerformanceUtils;
import flex.messaging.services.MessageService;
import flex.messaging.services.messaging.adapters.JMSConfigConstants;
import flex.messaging.services.messaging.adapters.MessagingAdapter;
import org.springframework.beans.MutablePropertyValues;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
import org.springframework.web.context.support.WebApplicationContextUtils;

import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
import java.util.Enumeration;
import java.util.UUID;


/**
 * Spring has the ability to configure and manage JMS message queues.
 * This adapater allows BlazeDS to use Spring to configure and connect to
 * different JMS queues
 *
 * @author Mike Nimer
 * @author Chris Scott
 * @todo - test dynamic topics and queues
 * @todo - make sure clients work with subtopics
 */
public class SpringJmsAdapter extends MessagingAdapter implements JmsConfigConstants, MessageListener
{
    // Adapter and Flex Destination properties
    private String destinationBeanName;
    private String jmsTemplateBeanName;
    private String jmsFactoryBeanName;


    //private String destinationType = defaultDestinationType; //TOPIC, QUEUE
    private String messageTypeClassName = TextMessage.class.getName();

    // local properties
    private ApplicationContext applicationContext;
    private DefaultMessageListenerContainer defaultListener;
    private ConnectionFactory jmsFactoryBean;
    private JmsTemplate jmsTemplateBean;
    private Destination destinationBean;


    /**
     * Initializes the properties.
     * <p/>
     * This is called 3 times. First the with Service Properties, then the Adapter properties, and finally with the Destination properties.
     * This allows you to have granular control overriding properties as you go down the configuration chain (for example, destination props override adapter props)     *
     *
     * @param properties Properties for the <code>SpringJmsAdapter</code>.
     */
    public void initialize(String id, ConfigMap properties)
    {
        super.initialize(id, properties);

        // properties.getProperty("hibernateFactory");
        ConfigMap jms = properties.getPropertyAsMap("jms", null);
        if (jms != null)
        {
            if (jms.containsKey(JMS_FACTORY_BEANNAME))
            {
                jmsFactoryBeanName = jms.getPropertyAsString(JMS_FACTORY_BEANNAME, null);
            }
            if (jms.containsKey(JMS_TEMPLATE_BEANNAME))
            {
                jmsTemplateBeanName = jms.getPropertyAsString(JMS_TEMPLATE_BEANNAME, null);
            }
            if (jms.containsKey(DESTINATION_BEANNAME))
            {
                destinationBeanName = jms.getPropertyAsString(DESTINATION_BEANNAME, null);
            }
            if (jms.containsKey(MESSAGE_TYPE))
            {
                messageTypeClassName = jms.getPropertyAsString(MESSAGE_TYPE, null);
            }
        }

    }

                                          
    /**
     * Validate the properties and construct the spring beans
     * it is started.
     */
    protected void validate()
    {
        if (isValid())
        {
            return;
        }

        super.validate();

        if (this.jmsTemplateBeanName == null)
        {
            ConfigurationException ce = new ConfigurationException();
            ce.setMessage("missing " + JMS_TEMPLATE_BEANNAME);
            throw ce;
        }
        if (this.destinationBeanName == null)
        {
            ConfigurationException ce = new ConfigurationException();
            ce.setMessage("missing " + DESTINATION_BEANNAME);
            throw ce;
        }
        if (this.jmsFactoryBeanName == null)
        {
            ConfigurationException ce = new ConfigurationException();
            ce.setMessage("missing " + JMS_FACTORY_BEANNAME);
            throw ce;
        }
        if (this.messageTypeClassName == null)
        {
            ConfigurationException ce = new ConfigurationException();
            ce.setMessage("missing " + MESSAGE_TYPE);
            throw ce;
        }

        // retrieve the beans we need from the application context
        retrieveSpringBeans();

        // construct a message listener container for the jms adapter
        createMessageListenerContainer();
    }


    /**
     * Starts the adapter
     */
    public void start()
    {
        super.start();
    }


    /**
     * Stops the adapter.
     */
    public void stop()
    {
        if (!isStarted())
        {
            return;
        }

        //TODO:  Tear down Spring Beans here, if needed

        super.stop();
    }


    public Object invoke(Message message)
    {
        MessageCreator msgCreator = new FlexMessageCreator(this.messageTypeClassName, message);
        return invoke(message, msgCreator);
    }


    public Object invoke(Message message, MessageCreator messageCreator)
    {
        try
        {
            jmsTemplateBean.setTimeToLive(getTimeToLive(message));
        } catch (JMSException ex)
        {
            jmsTemplateBean.setTimeToLive(60000);
        }

        jmsTemplateBean.send(destinationBean, messageCreator);

        return null;
    }


    public void onMessage(javax.jms.Message message)
    {
        System.out.println(message.toString());

        AsyncMessage flexMessage = jmsToFlexMessage(message);
        if (flexMessage != null)
        {
            MessageService msgService = (MessageService) getDestination().getService();
            msgService.pushMessageToClients(flexMessage, true);
        }
    }


    protected void retrieveSpringBeans()
    {
        applicationContext = WebApplicationContextUtils.getWebApplicationContext(FlexContext.getServletContext());
        jmsTemplateBean = (JmsTemplate) applicationContext.getBean(jmsTemplateBeanName);
        destinationBean = ((Destination) applicationContext.getBean(destinationBeanName));
        jmsFactoryBean = (ConnectionFactory) applicationContext.getBean(jmsFactoryBeanName);
    }


    /**
     * Constructs a message listener container bean and adds it to the default context. The JMS adapter
     * is registered with the message listener context bean as a message driven pojo.
     */
    protected void createMessageListenerContainer()
    {
        RootBeanDefinition beanDef = new RootBeanDefinition(DefaultMessageListenerContainer.class);
        MutablePropertyValues props = new MutablePropertyValues();
        props.addPropertyValue("connectionFactory", jmsFactoryBean);
        props.addPropertyValue("destination", destinationBean);
        props.addPropertyValue("messageListener", this);
        beanDef.setPropertyValues(props);

        // we need some kind of unique id for the beanName, I'll use a UUID
        String beanID = UUID.randomUUID().toString();

        // register then retrieve the new bean
        ConfigurableListableBeanFactory beanFactory = ((AbstractApplicationContext) applicationContext).getBeanFactory();
        ((DefaultListableBeanFactory) beanFactory).registerBeanDefinition(beanID, beanDef);

        defaultListener = (DefaultMessageListenerContainer) applicationContext.getBean(beanID);
    }


    protected AsyncMessage jmsToFlexMessage(javax.jms.Message jmsMessage)
    {
        flex.messaging.messages.AsyncMessage flexMessage = null;
        flexMessage = new AsyncMessage();

        //String clientId = (String) consumerToClientId.get(consumer);
        //flexMessage.setClientId(clientId);

        flexMessage.setDestination(getDestination().getId());

        try
        {
            // Set JMSMessageID header as Flex messageId property.
            flexMessage.setMessageId(jmsMessage.getJMSMessageID());
            flexMessage.setTimestamp(jmsMessage.getJMSTimestamp());

            // set JMS Header properties
            flexMessage.setHeader(JMS_CORRELATION_ID, jmsMessage.getJMSCorrelationID());
            flexMessage.setHeader(JMS_DELIVERY_MODE, Integer.toString(jmsMessage.getJMSDeliveryMode()));
            flexMessage.setHeader(JMS_DESTINATION, jmsMessage.getJMSDestination().toString());
            flexMessage.setHeader(JMS_EXPIRATION, Long.toString(jmsMessage.getJMSExpiration()));
            flexMessage.setHeader(JMS_PRIORITY, Integer.toString(jmsMessage.getJMSPriority()));
            flexMessage.setHeader(JMS_REDELIVERED, Boolean.toString(jmsMessage.getJMSRedelivered()));
            flexMessage.setHeader(JMS_REPLY_TO, jmsMessage.getJMSReplyTo());
            flexMessage.setHeader(JMS_TYPE, jmsMessage.getJMSType());
        }
        catch (JMSException jmsEx)
        {
            jmsEx.printStackTrace();
            // These should not cause errors to be pushed to Flash clients
            //Log.getLogger(LOG_CATEGORY).warn("JMSAdapter encountered an error while retrieving JMS headers during JMS to Flex conversion: " + jmsEx.getMessage());
        }


        // Set JMS properties as Flex headers.

        // While iterating through JMS message properties, build a message
        // performance object to send back to the client with the message
        // properties starting with with MPI_HEADER_IN (if any).
        MessagePerformanceInfo mpi = null;

        try
        {
            for (Enumeration propEnum = jmsMessage.getPropertyNames(); propEnum.hasMoreElements();)
            {
                String propName = (String) propEnum.nextElement();
                try
                {
                    Object propValue = jmsMessage.getObjectProperty(propName);
                    if (propName.startsWith(MessagePerformanceUtils.MPI_HEADER_IN))
                    {
                        if (mpi == null)
                        {
                            mpi = new MessagePerformanceInfo();
                        }
                        propName = propName.substring(MessagePerformanceUtils.MPI_HEADER_IN.length());
                        java.lang.reflect.Field field;
                        try
                        {
                            field = mpi.getClass().getField(propName);
                            field.set(mpi, propValue);
                        }
                        catch (Exception ignore)
                        {
                            // Simply don't set the property if the value cannot be retrieved.
                        }
                    } else
                    {
                        flexMessage.setHeader(propName, propValue);
                    }
                }
                catch (JMSException jmsEx)
                {
                    jmsEx.printStackTrace();
                    //Log.getLogger(LOG_CATEGORY).warn("JMSAdapter encountered an error while retrieving JMS properties during JMS to Flex conversion: " + jmsEx.getMessage());
                }
            }

            if (mpi != null)
            {
                flexMessage.setHeader(MessagePerformanceUtils.MPI_HEADER_IN, mpi);
            }
        }
        catch (JMSException jmsEx)
        {
            jmsEx.printStackTrace();
            //Log.getLogger(LOG_CATEGORY).warn("JMSAdapter encountered an error while retrieving JMS properties during JMS to Flex conversion: " + jmsEx.getMessage());
        }

        // Finally, set the JMS message body of the Flex message body.
        try
        {
            if (jmsMessage instanceof javax.jms.TextMessage)
            {
                javax.jms.TextMessage textMessage = (javax.jms.TextMessage) jmsMessage;
                flexMessage.setBody(textMessage.getText());
            } else if (jmsMessage instanceof javax.jms.ObjectMessage)
            {
                javax.jms.ObjectMessage objMessage = (javax.jms.ObjectMessage) jmsMessage;
                flexMessage.setBody(objMessage.getObject());
            }
        }
        catch (JMSException jmsEx)
        {
            jmsEx.printStackTrace();
        }

        return flexMessage;
    }


    protected long getTimeToLive(Message message) throws JMSException
    {
        long timeToLive = message.getTimeToLive();
        if (message.getHeaders().containsKey(JMSConfigConstants.TIME_TO_LIVE))
        {
            long l = ((Long) message.getHeader(JMSConfigConstants.TIME_TO_LIVE)).longValue();
            if (l != 0)
            {
                // Don't let Flex default override JMS implementation default,
                // only explicit ActionScript TTL usage overrides JMS TTL.
                timeToLive = l;
            }
        }
        return timeToLive;
    }


}
